package com.wzx.client.core;

import com.wzx.api.RpcRequest;
import com.wzx.api.Utils;
import com.wzx.client.ClientApplication;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

@Component
public class RpcClient {

    @Autowired
    private ApplicationContext applicationContext;

    static ExecutorService executor = Executors.newSingleThreadExecutor();

    private List<RpcNioSocketChannel> serverChannelList = new ArrayList<>();
    private int requestIndex = 0;

    @PostConstruct
    public void start() throws InterruptedException {
        for (int serverPort : ClientApplication.serverPorts) {
            RpcNioSocketChannel serverChannel = createServerChannel(serverPort);
            serverChannelList.add(serverChannel);
        }
    }

    RpcNioSocketChannel createServerChannel(int serverPort) throws InterruptedException {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);

        Bootstrap serverBootstrap = new Bootstrap()
                .group(bossGroup)
                .channel(RpcNioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        RpcClientHandler bean = applicationContext.getBean(RpcClientHandler.class);

                        ch.pipeline()
                                .addLast()
                                .addLast(new StringDecoder(), new StringEncoder())
                                .addLast(bean)
                        ;
                    }
                });

        ChannelFuture sync = serverBootstrap.connect("127.0.0.1", serverPort).sync();
        return (RpcNioSocketChannel) sync.channel();
    }

    public synchronized String request(RpcRequest rpcRequest) throws Exception {
        Future<String> submit = executor.submit(() -> {
            RpcNioSocketChannel channel = serverChannelList.get(requestIndex++ & serverChannelList.size() - 1);
            synchronized (channel) {
                String json = Utils.objectMapper.writeValueAsString(rpcRequest);
                channel.writeAndFlush(json);
                channel.wait();
                return channel.getResult();
            }
        });
        return submit.get();
    }
}
